package com.wuwangfu.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @Author jcshen
 * @Date 2023-02-21
 * @PackageName:com.wuwangfu.sink
 * @ClassName: DPrintSink
 * @Description: TODO
 * @Version 1.0
 * <p>
 *
 * print 并行度
 * localhost:8081
 */
public class DPrintSink {
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);

        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        System.out.println(line.getParallelism());

        //line.print();

        line.addSink(new MyPrintSink()).name("my-print-sink");
        env.execute();
    }

    /**
     * 自定义打印Sink
     */
    private static class MyPrintSink extends RichSinkFunction<String> {

        int idxSubTask;

        @Override
        public void open(Configuration parameters) throws Exception {
            idxSubTask = getRuntimeContext().getIndexOfThisSubtask();
        }

        @Override
        public void invoke(String value, Context context) throws Exception {
            System.out.println(idxSubTask + 1 + ">" + value);
        }
    }
}
